package feat1;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

public class SimplePublisher implements Publisher<String> {
    @Override
    public void subscribe(Subscriber<? super String> subscriber) {
        // 创建 Subscription 并传递给 Subscriber
        subscriber.onSubscribe(new Subscription() {
            @Override
            public void request(long n) {
                // 发送数据流
                subscriber.onNext("Hello");
                subscriber.onNext("Reactive");
                subscriber.onNext("Streams");
                subscriber.onComplete(); // 完成信号
            }

            @Override
            public void cancel() {
                // 取消逻辑（示例中未实现）
            }
        });
    }

    public static void main(String[] args) {
        SimplePublisher publisher = new SimplePublisher();

        // 创建 Subscriber
        Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onSubscribe(Subscription s) {
                s.request(3); // 请求3个元素
            }

            @Override
            public void onNext(String s) {
                System.out.println("Received: " + s);
            }

            @Override
            public void onError(Throwable t) {
                System.err.println("Error: " + t.getMessage());
            }

            @Override
            public void onComplete() {
                System.out.println("Data stream completed");
            }
        };

        publisher.subscribe(subscriber);
    }
}